package org.fastsyncer.manager.increment;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.fastsyncer.common.constant.ConnectorConstant;
import org.fastsyncer.common.constant.MappingConstant;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.manager.framework.ManagerFactory;
import org.fastsyncer.manager.increment.canel.CanelRemoteClient;
import org.fastsyncer.manager.increment.ldap.LdapListener;
import org.fastsyncer.manager.increment.logmnr.LogmnrListener;
import org.fastsyncer.manager.store.DataCenter;
import org.json.JSONArray;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class IncrementFactory extends Thread {

	IncrementFactory() {
		setName("fastsyncer increment listener thread");
	}
	
	private static Logger logger = LoggerFactory.getLogger(IncrementFactory.class);

	private static HashMap<String, Map<String, String>> RUNNING = new HashMap<String, Map<String, String>>();

	@Override
	public void run() {
		// 启动增量监听服务
		logger.info("Start incremental listening service...");
		while (true) {

			// 刷新同步列表任务
			this.refreshTask();

			for (Map.Entry<String, Map<String, String>> task : RUNNING.entrySet()) {
				try {
					JSONArray msg = null;
					Map<String, String> v = task.getValue();
					String type = v.get("type");
					if(StringUtils.isBlank(type)){
						continue;
					}
					switch (type) {
					case MappingConstant.POLICY_INCREMENT_MYSQL:
						// 如果是mysql增量配置
						msg = CanelRemoteClient.getInstance().pull(v);
						break;
					case MappingConstant.POLICY_INCREMENT_ORACLE:
						// 如果是oracle增量配置
						msg = LogmnrListener.getInstance().pull(v);
						break;
					case MappingConstant.POLICY_INCREMENT_LDAP:
						// 如果是目录服务增量配置
						msg = LdapListener.getInstance().pull(v);
						break;
					default:
						break;
					}
					// 将增量消息发送给manager处理
					if (msg != null) {
						JSONObject t = new JSONObject();
						t.put("taskId", v.get("taskId"));
						t.put("msg", msg);
						ManagerFactory.getInstance().handleSync(t);
					}
				} catch (Exception e) {
				    logger.error(e.getClass() + ">>" + e.getMessage() + ">>" + e.getStackTrace().toString());
				}
			}

			// 防止连接太频繁
			try {
				Thread.sleep(500);
			} catch (InterruptedException e) {
			}
		}
	}

	// 刷新同步列表任务
	public void refreshTask() {
		// 清空上次监听任务
		RUNNING.clear();
		List<Object> mappingAll = DataCenter.getInstance().getMappingList();
		for (Object object : mappingAll) {
			MappingTask task = (MappingTask) object;
			String connector = task.getSourceMapping().getConnector();
			// 排除WebService连接器
			if(ConnectorConstant.WS_CONNECTOR.equals(connector)){
			    continue;
			}
			
			if (!task.isEnable()) {
				continue;
			}
			// 读取增量同步任务
			boolean isIncrement = StringUtils.equals(MappingConstant.SYNC_INCREMENT, task.getModel());
			if (!isIncrement) {
				continue;
			}
			// 策略配置
			HashMap<String, HashMap<String, String>> policy = task.getPolicy();
			HashMap<String, String> po = policy.get(MappingConstant.POLICY_INCREMENT);
			if (po == null) {
				continue;
			}
			
			// 是否开启功能
			Boolean enable = Boolean.valueOf(po.get("enable"));
			if (enable) {
				po.put("taskId", task.getId());
				RUNNING.put(task.getId(), po);
			}
		}
	}

}
